package org.nbict.iot.trident.mogo.state;

import com.google.common.collect.Lists;
import org.apache.commons.lang.Validate;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.state.State;
import org.apache.storm.trident.tuple.TridentTuple;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.nbict.iot.trident.mogo.MongoDBClient;
import org.nbict.iot.trident.mogo.mapper.MongoMapper;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * mongo state for trident
 * Created by songseven on 18/6/25.
 */
public class MongoState implements State {

    private MongoDBClient mongoClient;
    private MongoState.Options options;
    private Map map;

    private Long txid;

    public MongoState(Map map, Options options) {
        this.map = map;
        this.options = options;
    }

    protected void prepare() {
        Validate.notEmpty(this.options.url, "url can not be blank or null");
        Validate.notEmpty(this.options.collectionName, "collectionName can " +
                "not be blank or null");
        Validate.notNull(this.options.mapper, "MongoMapper can not be null");

        this.mongoClient = new MongoDBClient(this.options.url, this.options.database, this.options.username, this.options.password, this
                .options.admin, this.options.collectionName);
    }

    @Override
    public void beginCommit(Long txid) {
        this.txid = txid;
    }

    @Override
    public void commit(Long aLong) {
    }

    /**
     * 更新值
     *
     * @param tuples
     * @param collector
     */
    public void updateState(List<TridentTuple> tuples, TridentCollector
            collector) {

        ArrayList documents = Lists.newArrayList();
        Iterator var4 = tuples.iterator();

        //System.out.println(tuples);

        while(var4.hasNext()) {
            TridentTuple tuple = null;
            Document document = null;

            tuple = (TridentTuple)var4.next();
            document = this.options.mapper.toDocument(tuple);

//            TridentTuple tuple = (TridentTuple)var4.next();
//            Document document = this.options.mapper.toDocument(tuple);
            Bson filter = this.options.mapper.filter(tuple); //更新值
            //System.out.println(filter); //debug
            //System.out.println(document);//debug
            if (filter != null) {
                mongoClient.update(filter, document, true); //没有则更新
            } else {
                documents.add(document);
            }
        }
        if (!documents.isEmpty()) {
            this.mongoClient.insert(documents, true); //插入值 ordered
        }

    }

    public static class Options implements Serializable {
        private String url;
        private String database;
        private String username;
        private String password;
        private String admin;
        private String collectionName;
        private MongoMapper mapper;

        public Options() {
        }

        public MongoState.Options withUrl(String url) {
            this.url = url;
            return this;
        }

        public MongoState.Options withDatabase(String database) {
            this.database = database;
            return this;
        }

        public MongoState.Options withUsername(String username) {
            this.username = username;
            return this;
        }

        public MongoState.Options withPassword(String password) {
            this.password = password;
            return this;
        }

        public MongoState.Options withAdmin(String admin) {
            this.admin = admin;
            return this;
        }

        public MongoState.Options withCollectionName(String collectionName) {
            this.collectionName = collectionName;
            return this;
        }

        public MongoState.Options withMapper(MongoMapper mapper) {
            this.mapper = mapper;
            return this;
        }

    }
}
